package com.hisense.ovcloud.ps.service;

import cn.hutool.json.JSONUtil;
import com.hisense.ovcloud.commons.configs.base.BaseResponse;
import com.hisense.ovcloud.commons.configs.kafka.ReactiveKafkaSender;
import com.hisense.ovcloud.commons.configs.redis.RedisDao;
import com.hisense.ovcloud.ps.Constants;
import com.hisense.ovcloud.ps.config.WorkPoolProperties;
import com.hisense.ovcloud.ps.dto.common.OnAndOfflineDataToKafka;
import com.hisense.ovcloud.ps.dto.common.WgpsStatus;
import com.hisense.ovcloud.ps.dto.common.WifiStatsFromRedis;
import com.hisense.ovcloud.ps.dto.req.GetWifiListStatReq;
import com.hisense.ovcloud.ps.dto.req.GetWifiStatReq;
import com.hisense.ovcloud.ps.dto.req.WifiOnlineReq;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.validation.annotation.Validated;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.util.List;

@Service
@Slf4j
@Validated
public class PsService implements InitializingBean {
    private final RedisDao redisDao;

    private final ReactiveKafkaSender reactiveKafkaSender;

    private final WorkPoolProperties workPoolProperties;

    private static Scheduler kafkaWorkPool;
    private static Scheduler redisReaderPool;
    private static Scheduler redisWriterPool;


    @Value("${wifiIdTTL:200}")
    private int wifiIdTTL;

    public PsService(RedisDao redisDao, ReactiveKafkaSender reactiveKafkaSender, WorkPoolProperties workPoolProperties) {
        this.redisDao = redisDao;
        this.reactiveKafkaSender = reactiveKafkaSender;
        this.workPoolProperties = workPoolProperties;
    }

    @Override
    public void afterPropertiesSet() {
        kafkaWorkPool = Schedulers.newBoundedElastic(workPoolProperties.getKafkaPoolCap(), workPoolProperties.getKafkaPoolQueuelength(), "kafka-work-pool");
        redisReaderPool = Schedulers.newBoundedElastic(workPoolProperties.getRedisReaderPoolCap(), workPoolProperties.getRedisReaderQueuelength(), "redis-reader-pool");
        redisWriterPool = Schedulers.newBoundedElastic(workPoolProperties.getRedisWriterPoolCap(), workPoolProperties.getRedisWriterQueuelength(), "redis-writer-pool");
    }


    public Mono<BaseResponse> wifiOnline(Mono<WifiOnlineReq> req) {
        return req.flatMap(r -> getWifiStatFromRedis(r.getWifiId(), Mono.just(r.getWgpsStatus()))
                .zipWhen(wifiStatsFromRedis -> {
                    if (wifiStatsFromRedis.isFlag()) {
                        return onLineOperateRedis(wifiStatsFromRedis, Mono.just(r.getWgpsStatus())).zipWhen(
                                result -> {
                                    if (result) {
                                        String timeStampStr = String.valueOf(System.currentTimeMillis());
                                        reactiveKafkaSender.sendToKafkaReactiveShard(Constants.WIFIOnline.KAFKA_TOPIC_ONLINE_STATUS,
                                                r.getWifiId(),
                                                JSONUtil.toJsonStr(OnAndOfflineDataToKafka.builder()
                                                        .wifiId(r.getWifiId())
                                                        .wifiStatus(String.valueOf(r.getWgpsStatus().getStat()))
                                                        .netId(String.valueOf(r.getWgpsStatus().getNetId()))
                                                        .wifiVersion(r.getWgpsStatus().getWifiver())
                                                        .ownerServer(r.getWgpsStatus().getOwner())
                                                        .devVer(r.getWgpsStatus().getDevVer())
                                                        .peerAddr(r.getWgpsStatus().getPeeraddr())
                                                        .time(timeStampStr)
                                                        .statusTime(timeStampStr)
                                                        .build()), kafkaWorkPool);
                                    }
                                    return Mono.just(true);
                                }
                                , (x, y) -> y);
                    } else {
                        return Mono.just(false);
                    }
                }, (u, v) -> v)).flatMap(result -> Mono.just(BaseResponse.builder().data(result).build()));
    }

    public Mono<BaseResponse> getWifiStats(Mono<GetWifiStatReq> req) {
        return getWifiStatsBase(req).zipWhen(wgpsStatus -> Mono.just(BaseResponse.builder().data(wgpsStatus).build()), (x, y) -> y);
    }

    public Mono<WgpsStatus> getWifiStatsBase(Mono<GetWifiStatReq> req) {
        return req.flatMap(r -> redisDao.get(r.getWifiId())
                .switchIfEmpty(Mono.just(Constants.REDIS_NULL_VALUE))
                .zipWhen(valInRedis -> {
                    if (valInRedis.equals(Constants.REDIS_NULL_VALUE)) {
                        return Mono.just(WgpsStatus.builder().wifiId(r.getWifiId()).stat(Constants.WIFIOnline.OFFLINE_STATUS).build());
                    } else {
                        return Mono.just(JSONUtil.toBean(valInRedis, WgpsStatus.class));
                    }
                }, (u, v) -> v));
    }


    public Mono<BaseResponse> getWifiListStats(Mono<GetWifiListStatReq> req) {
        return getWifiListStatsBase(req).zipWhen(list -> Mono.just(BaseResponse.builder().data(list).build()), (x, y) -> y);
    }

    public Mono<List<WgpsStatus>> getWifiListStatsBase(Mono<GetWifiListStatReq> req) {
        return req.flatMap(r -> Flux.fromIterable(r.getWifiList()).flatMap(wifiId -> redisDao.get(wifiId)
                .switchIfEmpty(Mono.just(Constants.REDIS_NULL_VALUE))
                .zipWhen(redisVal -> {
                    if (redisVal.equals(Constants.REDIS_NULL_VALUE)) {
                        return Mono.just(WgpsStatus.builder().wifiId(wifiId).stat(Constants.WIFIOnline.OFFLINE_STATUS).wifiId(wifiId).build());
                    } else {
                        return Mono.just(JSONUtil.toBean(redisVal, WgpsStatus.class));
                    }
                }, (u, v) -> v).subscribeOn(redisReaderPool)).collectList());
    }

    public Mono<WifiStatsFromRedis> getWifiStatFromRedis(String wifiId, Mono<WgpsStatus> req) {
        return req.flatMap(r -> redisDao.get(wifiId)
                .subscribeOn(redisReaderPool)
                .switchIfEmpty(Mono.just(Constants.REDIS_NULL_VALUE))
                .flatMap(redisVal -> {
                    if (r.getStat() != Constants.WIFIOnline.OFFLINE_STATUS) {
                        return Mono.just(WifiStatsFromRedis.builder().flag(true).data(redisVal).build());
                    }
                    if (redisVal.equals(Constants.REDIS_NULL_VALUE)) {
                        return Mono.just(WifiStatsFromRedis.builder().flag(true).data(Constants.REDIS_NULL_VALUE).build());
                    }
                    if (r.getNetId() == JSONUtil.toBean(redisVal, WgpsStatus.class).getNetId() &&
                            r.getOwner().equals(JSONUtil.toBean(redisVal, WgpsStatus.class).getOwner())) {
                        return Mono.just(WifiStatsFromRedis.builder().flag(true).data(redisVal).build());
                    } else {
                        return Mono.just(WifiStatsFromRedis.builder().flag(false).build());
                    }
                }));
    }


    public Mono<Boolean> onLineOperateRedis(WifiStatsFromRedis wifiStatsFromRedis, Mono<WgpsStatus> req) {
        return req.flatMap(r -> Mono.just(wifiStatsFromRedis.getData())
                .flatMap(redisVal -> {
                    String peerAddr = r.getPeeraddr();
                    if (redisVal.equals(Constants.REDIS_NULL_VALUE)) {
                        r.setPeeraddr("");
                        return redisDao.set(r.getWifiId(), JSONUtil.toJsonStr(r), wifiIdTTL)
                                .subscribeOn(redisWriterPool)
                                .zipWhen(succ -> {
                                    r.setPeeraddr(peerAddr);
                                    return Mono.just(true);
                                }, (u, v) -> v);
                    } else {
                        if (redisVal.equals(JSONUtil.toJsonStr(r))) {
                            return redisDao.set(r.getWifiId(), redisVal, wifiIdTTL).flatMap(res -> Mono.just(false));
                        } else {
                            boolean flag101 = true;
                            if (StringUtils.isEmpty(r.getWifiver())) {
                                flag101 = false;
                                r.setWifiver(JSONUtil.toBean(redisVal, WgpsStatus.class).getWifiver());
                            }
                            r.setPeeraddr("");
                            final boolean flagUsed = flag101;
                            return redisDao.set(r.getWifiId(), JSONUtil.toJsonStr(r), wifiIdTTL)
                                    .subscribeOn(redisWriterPool)
                                    .zipWhen(succ -> {
                                        r.setPeeraddr(peerAddr);
                                        if (!flagUsed && r.getStat() == JSONUtil.toBean(redisVal, WgpsStatus.class).getStat()) {
                                            return Mono.just(false);
                                        }
                                        return Mono.just(true);
                                    }, (u, v) -> v);
                        }
                    }
                }));
    }
}
